package com.shujia.flink.sql

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object Demo14CityFlow {
  def main(args: Array[String]): Unit = {

    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .inStreamingMode()
      .build()
    /**
     * flink sql环境
     *
     */
    val table: TableEnvironment = TableEnvironment.create(settings)


    table.executeSql(
      """
        |
        |CREATE TABLE cars (
        |    car STRING,
        |    city_code STRING,
        |    county_code STRING,
        |    card BIGINT,
        |    camera_id STRING,
        |    orientation STRING,
        |    road_id BIGINT,
        |    `time` STRING,
        |    speed DOUBLE
        |)  WITH (
        |  'connector' = 'filesystem',           -- 必选：指定连接器类型
        |  'path' = 'data/cars_sample.json',  -- 必选：指定路径
        |  'format' = 'json'                     -- 必选：文件系统连接器指定 format
        |)
        |
        |
        |""".stripMargin)

    table.executeSql(
      """
        |
        |CREATE TABLE city_flow (
        |  city_code STRING,
        |  flow BIGINT,
        |  PRIMARY KEY (city_code) NOT ENFORCED -- 按照主键更新数据
        |) WITH (
        |   'connector' = 'jdbc',
        |   'url' = 'jdbc:mysql://master:3306/bigdata?useUnicode=true&characterEncoding=UTF-8',
        |   'table-name' = 'city_flow', -- 需要手动到数据库中创建表
        |   'username' = 'root',
        |   'password' = '123456'
        |)
        |""".stripMargin)


    table.executeSql(
      """
        |
        |insert into city_flow
        |select
        |city_code,
        |count(distinct car) as flow
        |from
        |cars
        |group by city_code
        |
        |""".stripMargin)

  }

}
